package com.shujia.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;

public class Demo03Filter {
    Connection conn;
    Admin admin; // 对元数据进行操作：create、drop、list、alter、disable、enable
    Table table; // 先通过conn.getTable来初始化，对表的数据进行操作

    @Before
    public void init() throws IOException {
        Configuration conf = new Configuration();
        // 指定HBase地址
        conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");
        // 创建HBase连接
        conn = ConnectionFactory.createConnection(conf);
        admin = conn.getAdmin();

    }

    // 过滤出文科班的男同学
    @Test
    public void filter01() throws IOException {
        Table students = conn.getTable(TableName.valueOf("students"));

        Scan scan = new Scan();

        // 使用列值过滤器
        SingleColumnValueFilter filter1 = new SingleColumnValueFilter(
                "info".getBytes()
                , "clazz".getBytes()
                , CompareOperator.EQUAL
                , new BinaryPrefixComparator("文科".getBytes())
        );

        SingleColumnValueFilter filter2 = new SingleColumnValueExcludeFilter(
                "info".getBytes()
                , "gender".getBytes()
                , CompareOperator.EQUAL
                , new BinaryComparator("男".getBytes())
        );

        // FilterList
        FilterList filterList = new FilterList(
                FilterList.Operator.MUST_PASS_ALL
                , filter1
                , filter2
        );

        scan.setFilter(filterList);

        ResultScanner scanner = students.getScanner(scan);
        for (Result rs : scanner) {
            String id = Bytes.toString(rs.getRow());
            String name = Bytes.toString(rs.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(rs.getValue("info".getBytes(), "age".getBytes()));
            String gender = "男";
            String clazz = Bytes.toString(rs.getValue("info".getBytes(), "clazz".getBytes()));
            System.out.println(id + "," + name + "," + age + "," + gender + "," + clazz);
        }
    }


    @After
    public void close() throws IOException {
        // 关闭连接
        conn.close();
    }
}
